# -*- coding: utf-8 -*-
from datetime import timedelta
from time import time
from utils.operators.rich_sql_sensor import RichSqlSensor

label = f"dwd_work_order_new_base_dt_{int(time())}"
timeout = timedelta(minutes=10).seconds
doris_jms_dwd__dwd_work_order_new_base_dt = RichSqlSensor(
    task_id='doris_jms_dwd__dwd_work_order_new_base_dt',
    email=['shenjiaming@jtexpress.com', 'yl_bigdata@yl-scm.com'],
    pool='broker_load_pool',
    conn_id='new_doris',
    pre_sql=f"""
             TRUNCATE TABLE jms_dwd.dwd_work_order_new_base_dt PARTITION (p{{{{ execution_date | cst_ds_nodash }}}});
             LOAD LABEL jms_dwd.{label} (
                 DATA INFILE("hdfs://{{{{var.value.hadoop_namespace}}}}/dw/hive/jms_dwd.db/external/dwd_work_order_new_base_dt/dt={{{{ execution_date | cst_ds }}}}/*") 
                 INTO TABLE dwd_work_order_new_base_dt
                    FORMAT AS 'PARQUET'(
                    id,work_order_no,customer_name,customer_sex,customer_phone,customer_type,receiver_province_name,receiver_province_id,receiver_city_name,receiver_city_id,receiver_area_name,receiver_area_id,receiver_detailed_address,waybill_no,first_type_id,first_type_code,first_type_name,second_type_id,second_type_code,second_type_name,second_type_reminder_count,second_type_treatment_limitation,problem_description,emergency_level,accept_network_id,accept_network_code,accept_network_name,accept_network_type_id,call_back_name,call_back_phone,reminder_count,upgrade_status,upgrade_count,upgrade_path, status,source_code,registration_network_id,registration_network_code,registration_network_name,registration_network_type_id,accept_by,accept_by_code,accept_by_name,is_enable,is_delete,create_by,update_by,create_by_code,update_by_code,create_by_name,update_by_name,create_time,update_time,phone_count,accept_time,response_time,upgrade_time,sound_record_id,total_duration,duty_network_id,duty_network_code,duty_network_name,duty_network_type_id,record_group,process_status,service_code,service_name,service_registrant_code,service_registrant_name,old_accept_network_id,old_accept_network_code,old_accept_network_name,old_accept_network_type_id,accept_belong_network_id,accept_belong_network_code,accept_belong_network_name,accept_belong_network_type_id,response_state,transfer_network_id,transfer_network_code,transfer_network_name,transfer_network_type_id,monitoring_time,is_repeat,accept_network_logo,accept_league_network_id,reg_league_network_id,reg_network_logo,reg_belong_network_id,accept_franchisee_network_id,accept_franchisee_network_code,accept_franchisee_network_name,response_network_id,response_network_code,response_network_name,transfer_network_type,delivery_code,delivery_name,user_code,user_name,accept_belong_mr_code,accept_belong_mr_name,order_source_name,order_source_code,is_call,call_num,member_type,order_id,provider_id,provider_desc,city_id,city_desc,reopen,pick_network_code,pick_network_name,incoming_time,send_type,send_type_address,is_dispatcher,transfer_agent_network_code,transfer_agent_network_name,terminal_dispatch_code,is_vip
                    )
                    columns from path as (dt)
                    set (
                    id = id,work_order_no = work_order_no,customer_name = customer_name,customer_sex = customer_sex,customer_phone = customer_phone,customer_type = customer_type,receiver_province_name = receiver_province_name,receiver_province_id = receiver_province_id,receiver_city_name = receiver_city_name,receiver_city_id = receiver_city_id,receiver_area_name = receiver_area_name,receiver_area_id = receiver_area_id,receiver_detailed_address = receiver_detailed_address,waybill_no = waybill_no,first_type_id = first_type_id,first_type_code = first_type_code,first_type_name = first_type_name,second_type_id = second_type_id,second_type_code = second_type_code,second_type_name = second_type_name,second_type_reminder_count = second_type_reminder_count,second_type_treatment_limitation = second_type_treatment_limitation,problem_description = problem_description,emergency_level = emergency_level,accept_network_id = accept_network_id,accept_network_code = accept_network_code,accept_network_name = accept_network_name,accept_network_type_id = accept_network_type_id,call_back_name = call_back_name,call_back_phone = call_back_phone,reminder_count = reminder_count,upgrade_status = upgrade_status,upgrade_count = upgrade_count,upgrade_path = upgrade_path, status = status,source_code = source_code,registration_network_id = registration_network_id,registration_network_code = registration_network_code,registration_network_name = registration_network_name,registration_network_type_id = registration_network_type_id,accept_by = accept_by,accept_by_code = accept_by_code,accept_by_name = accept_by_name,is_enable = is_enable,is_delete = is_delete,create_by = create_by,update_by = update_by,create_by_code = create_by_code,update_by_code = update_by_code,create_by_name = create_by_name,update_by_name = update_by_name,create_time = create_time,update_time = update_time,phone_count = phone_count,accept_time = accept_time,response_time = response_time,upgrade_time = upgrade_time,sound_record_id = sound_record_id,total_duration = total_duration,duty_network_id = duty_network_id,duty_network_code = duty_network_code,duty_network_name = duty_network_name,duty_network_type_id = duty_network_type_id,record_group = record_group,process_status = process_status,service_code = service_code,service_name = service_name,service_registrant_code = service_registrant_code,service_registrant_name = service_registrant_name,old_accept_network_id = old_accept_network_id,old_accept_network_code = old_accept_network_code,old_accept_network_name = old_accept_network_name,old_accept_network_type_id = old_accept_network_type_id,accept_belong_network_id = accept_belong_network_id,accept_belong_network_code = accept_belong_network_code,accept_belong_network_name = accept_belong_network_name,accept_belong_network_type_id = accept_belong_network_type_id,response_state = response_state,transfer_network_id = transfer_network_id,transfer_network_code = transfer_network_code,transfer_network_name = transfer_network_name,transfer_network_type_id = transfer_network_type_id,monitoring_time = monitoring_time,is_repeat = is_repeat,accept_network_logo = accept_network_logo,accept_league_network_id = accept_league_network_id,reg_league_network_id = reg_league_network_id,reg_network_logo = reg_network_logo,reg_belong_network_id = reg_belong_network_id,accept_franchisee_network_id = accept_franchisee_network_id,accept_franchisee_network_code = accept_franchisee_network_code,accept_franchisee_network_name = accept_franchisee_network_name,response_network_id = response_network_id,response_network_code = response_network_code,response_network_name = response_network_name,transfer_network_type = transfer_network_type,delivery_code = delivery_code,delivery_name = delivery_name,user_code = user_code,user_name = user_name,accept_belong_mr_code = accept_belong_mr_code,accept_belong_mr_name = accept_belong_mr_name,order_source_name = order_source_name,order_source_code = order_source_code,is_call = is_call,call_num = call_num,member_type = member_type,order_id = order_id,provider_id = provider_id,provider_desc = provider_desc,city_id = city_id,city_desc = city_desc,reopen = reopen,pick_network_code = pick_network_code,pick_network_name = pick_network_name,incoming_time = incoming_time,send_type = send_type,send_type_address = send_type_address,is_dispatcher = is_dispatcher,transfer_agent_network_code = transfer_agent_network_code,transfer_agent_network_name = transfer_agent_network_name,terminal_dispatch_code = terminal_dispatch_code,is_vip = is_vip,date_time = dt
                    )
                    ) 
             WITH BROKER '{{{{ var.json.doris_brokers | random_choice }}}}'
             PROPERTIES ('timeout'='{timeout}', 'max_filter_ratio'='0.0')""",
    poke_sql=f"SHOW LOAD FROM jms_dwd WHERE label = '{label}' ORDER BY CreateTime DESC LIMIT 1",
    sql_on_kill=f"CANCEL LOAD FROM jms_dwd WHERE LABEL = '{label}'",
    success=lambda r: r[2] == 'FINISHED',
    failure=lambda r: (r[2] is not None and r[2] == 'CANCELLED', str(r[7])),
    poke_interval=60,
    # execution_timeout=max(datetime.now().replace(hour=8, minute=20, second=0) - datetime.now(), timedelta(seconds=1)),
)

